package david.java.flink_sql.test.leftjoin;


import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.IntStream;
/**
 * @Description:
 * @Author: ZhaoDawei
 * @Date: Create in 7:36 下午 2022/10/20
 */
public class Demo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, bsSettings);
        env.setParallelism(1);

        DataStream<Waybill> waybillDataStream = env.addSource(new CommonRealtimeMock<>(Arrays.asList(
                new Waybill(1, 2, 0, 1),
                new Waybill(1, 2, 30, 2),
                new Waybill(1, 2, 50, 3))
                , Waybill.class, 1)
        ).returns(Waybill.class);
        DataStream<Package> packageDataStream = env.addSource(new CommonRealtimeMock<>(Arrays.asList(
                new Package(2, "x", 0),
                new Package(2, "y", 4)
        ), Package.class,1)).returns(Package.class);

        tableEnv.createTemporaryView("waybill", waybillDataStream, "id,bm_pkg_id,status,utime");
        tableEnv.createTemporaryView("package", packageDataStream, "id,ext,utime");

        tableEnv.registerFunction("LastVal", new LatestIntAggregateFunction());

        RowToJsonSerializationSchema rowToJsonSerializationSchema = new RowToJsonSerializationSchema(Arrays.asList("waybill_id", "waybill_status", "package_ext"));
        String sql = "select waybill_id,\n" +
                "       waybill_status,\n" +
                "       package_ext\n" +
                "  from (\n" +
                "        select id waybill_id,\n" +
                "               max(bm_pkg_id) bm_pkg_id,\n" +
                "               LastVal(waybill.status,utime) waybill_status\n" +
                "          from waybill\n" +
                "         group by id\n" +
                "       ) w\n" +
                "  left join (\n" +
                "        select *\n" +
                "          from (\n" +
                "                select id,\n" +
                "                       ext package_ext,\n" +
                "                       row_number() over(partition by id order by utime desc) as rno\n" +
                "                  from package\n" +
                "               ) l\n" +
                "         where l.rno = 1\n" +
                "       ) p\n" +
                "    on w.bm_pkg_id = p.id";

        System.out.println(sql);
        Table result = tableEnv.sqlQuery(sql);
        DataStream<Tuple2<Boolean, Row>> stream = tableEnv.toRetractStream(result, Row.class);
        SingleOutputStreamOperator<Row> rows = stream.flatMap(new FlatMapFunction<Tuple2<Boolean, Row>, Row>() {
            @Override
            public void flatMap(Tuple2<Boolean, Row> row, Collector<Row> collector) throws Exception {
                if (row.f0) {
                    System.err.println("offset " + Util.outputcount.addAndGet(1) + " : " + new String(rowToJsonSerializationSchema.serialize(row.f1)));
                }
            }
        });
        env.execute();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Waybill implements Serializable {
        private long id;
        private long bm_pkg_id;
        private int status;
        private int utime;
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Package implements Serializable {
        private long id;
        private String ext;
        private int utime;
    }

    public static class RowToJsonSerializationSchema implements SerializationSchema<Row> {

        private final List<String> sqlResultFields;

        public RowToJsonSerializationSchema(List<String> sqlResultFields) {
            this.sqlResultFields = sqlResultFields;
        }

        @Override
        public byte[] serialize(Row row) {
            Map<String, Object> jsonMap = new HashMap<>();
            IntStream.range(0, row.getArity()).forEach(pos -> {
                final String fieldName = sqlResultFields.get(pos);
                final Object value = row.getField(pos);
                if (value == null) {
                    return;
                }
                jsonMap.put(fieldName, row.getField(pos));

            });
            return JSONObject.toJSONString(jsonMap).getBytes();
        }
    }

    public static class LatestIntAggregateFunction extends AggregateFunction<Integer, LatestIntAggregateFunction.LatestAccumulator> {
        @Override
        public LatestIntAggregateFunction.LatestAccumulator createAccumulator() {
            return new LatestIntAggregateFunction.LatestAccumulator();
        }

        @Override
        public Integer getValue(LatestIntAggregateFunction.LatestAccumulator accumulator) {
            return accumulator.getValue();
        }

        public void accumulate(LatestIntAggregateFunction.LatestAccumulator accumulator, Integer value, Integer order) {
            if (order == null) {
                return;
            }
            if (accumulator.getOrder() == null || order >= accumulator.getOrder()) {
                accumulator.setValue(value);
                accumulator.setOrder(order);
            }
        }

        public void retract(LatestIntAggregateFunction.LatestAccumulator accumulator, Integer value, int order) {

        }

        public void merge(LatestIntAggregateFunction.LatestAccumulator acc, Iterable<LatestIntAggregateFunction.LatestAccumulator> it) {
            for (LatestIntAggregateFunction.LatestAccumulator a : it) {
                if (a.getOrder() == null) {
                    continue;
                }
                if (acc.getOrder() == null || a.getOrder() >= acc.getOrder()) {
                    acc.setOrder(a.getOrder());
                    acc.setValue(a.getValue());
                }
            }
        }

        public void resetAccumulator(LatestIntAggregateFunction.LatestAccumulator acc) {
            acc.setValue(null);
            acc.setOrder(Integer.MIN_VALUE);
        }

        public static class LatestAccumulator {

            private Integer value;
            private Integer order;

            public Integer getValue() {
                return value;
            }

            public void setValue(Integer value) {
                this.value = value;
            }

            public Integer getOrder() {
                return order;
            }

            public void setOrder(Integer order) {
                this.order = order;
            }
        }
    }

    /**
     * 根据时间字段 ，模拟真实的处理间隔
     *
     * @param <T>
     */
    public static class CommonRealtimeMock<T> implements SourceFunction<T> {
        private static final long serialVersionUID = 1L;
        private Integer startDelay = 1;
        private Class clazz;
        private String timeField = "utime";
        private int timeUnit = 1000;
        private boolean print = true;

        @Setter
        private boolean sleep = true;

        public CommonRealtimeMock(List<T> datas, Class<T> clazz, int startDelay) {
            assert CollectionUtils.size(datas) > 0;
            this.startDelay = startDelay;
            this.clazz = clazz;
            data = datas;
        }

        private volatile boolean isRunning = true;

        List<T> data;

        @Override
        public void run(SourceContext<T> ctx) throws Exception {
            T temp = data.get(0);
            Integer time = Util.getInt(Util.getFiledIntValueNoException(temp, clazz, timeField));
            if (sleep) {
                Thread.sleep((long) startDelay * timeUnit);
            }
            ctx.collect(temp);
            if (print) {
                System.out.println(Util.format(System.currentTimeMillis()) + ": " + temp);
            }
            int counter = 1;
            while (isRunning && counter < data.size()) {
                temp = data.get(counter);
                Integer nextTime = Util.getInt(Util.getFiledIntValueNoException(temp, clazz, timeField));
                if (sleep) {
                    Thread.sleep((long) (nextTime - time) * timeUnit);
                }
                time = nextTime;
                ctx.collect(temp);
                if (print) {
                    System.out.println(Util.format(System.currentTimeMillis()) + ": " + temp);
                }
                counter++;
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }

}
